package com.frank.demo.ch02;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.omg.SendingContext.RunTime;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class HelloWorldConsumer {
    private volatile  boolean keepConsuming = true;

    public static void main(String[] args) {
        Properties kaProp = new Properties();
        kaProp.put("bootstrap.servers", "192.168.56.3:9000,192.168.56.3:9001,192.168.56.3:9002");
        kaProp.put("group.id", "helloWorldConsumer");
        kaProp.put("enable.auto.commit", "true");
        kaProp.put("auto.commit.interval.ms", "1000");
        kaProp.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        kaProp.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        HelloWorldConsumer consumer = new HelloWorldConsumer();
        consumer.consume(kaProp);

        Runtime.getRuntime().addShutdownHook(new Thread(consumer::shutdown));
    }

    private void consume(Properties kaProp){
        try(KafkaConsumer<String, String> consumer = new KafkaConsumer<>(kaProp)){
            consumer.subscribe(Arrays.asList("helloworld"));

            while (keepConsuming){
                System.out.println(LocalDateTime.now());
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(2000));
                for(ConsumerRecord<String, String> record : records){
                    String msg = String.format("helloWorld info: offset=%s, key=%s, value=%s", record.offset(), record.key(), record.value());
                    System.out.println(msg);
                }
            }
        }
    }
    private void shutdown(){
        keepConsuming = false;
    }
}
